#include <basic/errcode.h>
#include <basic/datetime.h>
#include <basic/list.h>
#include <basic/thrd.h>

int thrd_create(THRD_FUNC_T func, void *arg, unsigned int flag,
        unsigned long *id)
{/*{{{*/
    pthread_attr_t thrd_attr;
    int stack_size;
    pthread_t thrd_id;

    if(!func)
        return(RET_ERR_PARA);

    pthread_attr_init(&thrd_attr);
    if(flag & THRD_DETACH)
        pthread_attr_setdetachstate(&thrd_attr,
                PTHREAD_CREATE_DETACHED);
    if(flag & THRD_16K_STACK)
        stack_size=KB(16);
    else if(flag & THRD_128K_STACK)
        stack_size=KB(128);
    else if(flag & THRD_256K_STACK)
        stack_size=KB(256);
    else if(flag & THRD_512K_STACK)
        stack_size=KB(512);
    else
        stack_size=KB(64);
    pthread_attr_setstacksize(&thrd_attr, stack_size);
    if(pthread_create(&thrd_id, &thrd_attr, func, (void *)arg))
        return(-1);
    pthread_attr_destroy(&thrd_attr);
    if(id)
        *id=(unsigned long)thrd_id;

    return(0);
}/*}}}*/

int thrd_join(unsigned long thrd_id)
{/*{{{*/
    return(pthread_join((pthread_t)thrd_id, NULL));
}/*}}}*/

struct thrd_msgq
{/*{{{*/
    pthread_mutex_t mutex;
    pthread_cond_t cond;
    struct listp_link head;
    int count;
};/*}}}*/

int thrd_msgq_create(THRD_MSGQ *pmsgq)
{/*{{{*/
    THRD_MSGQ msgq;

    if(!pmsgq)  return(RET_ERR_PARA);
    *pmsgq=NULL;
    ALLOC_RETERR(msgq, struct thrd_msgq, RET_ERR_ALLOC);
    pthread_mutex_init(&msgq->mutex, NULL);
    pthread_cond_init(&msgq->cond, NULL);
    listp_init(&msgq->head); 
    msgq->count=0;
    *pmsgq=msgq;
    return(0);
}/*}}}*/

int thrd_msgq_destroy(THRD_MSGQ msgq)
{/*{{{*/
    if(!msgq)  return(RET_ERR_PARA);
    pthread_mutex_destroy(&msgq->mutex);
    pthread_cond_destroy(&msgq->cond);
    free(msgq);
    return(0);
}/*}}}*/

static int64_t _begin(int timeout_ms, struct timespec *ts)
{/*{{{*/
    struct timeval tv;
    gettimeofday(&tv, NULL);
    int64_t begin=tv.tv_sec*1000LL+tv.tv_usec/1000;
#if __WORDSIZE < 64
    ts->tv_sec=tv.tv_sec+timeout_ms/1000+1;
    ts->tv_nsec=0;
#else
    ts->tv_sec=tv.tv_sec + timeout_ms/1000;
    ts->tv_nsec=(tv.tv_usec + timeout_ms%1000*1000L) * 1000L;
    if(ts->tv_nsec > 1000000000L)
    {
        ts->tv_sec+=1;
        ts->tv_nsec-=1000000000L;
    }
#endif
    return(begin);
}/*}}}*/

static int _left(int timeout_ms, int64_t begin)
{/*{{{*/
    struct timeval tv;
    gettimeofday(&tv, NULL);
    int used=tv.tv_sec*1000LL+tv.tv_usec/1000-begin;
    return(timeout_ms > used ? timeout_ms-used : 0);
}/*}}}*/

int thrd_msgq_recv(THRD_MSGQ msgq, struct listp_link *list_recv,
        int timeout_ms)
{/*{{{*/
    if(!list_recv) return(RET_ERR_PARA);
    listp_init(list_recv);
    if(!msgq) return(RET_ERR_PARA);

    pthread_mutex_lock(&msgq->mutex);

    if(timeout_ms >= 0)
    {
        struct timespec ts;
        int64_t begin=_begin(timeout_ms, &ts);
        while(listp_is_empty(&msgq->head))
        {
            begin=_begin(timeout_ms, &ts);
            if(timeout_ms == 0 ||
                    pthread_cond_timedwait(&msgq->cond,
                        &msgq->mutex, &ts) == ETIMEDOUT)
            {
                pthread_mutex_unlock(&msgq->mutex);
                return(RET_ERR_TIMEOUT);
            }
            timeout_ms=_left(timeout_ms, begin);
        }
    }
    else
    {
        while(listp_is_empty(&msgq->head))
            pthread_cond_wait(&msgq->cond, &msgq->mutex);
    }
    listp_takeover(&msgq->head, list_recv);
    msgq->count=0;

    pthread_mutex_unlock(&msgq->mutex);

    return(0);
}/*}}}*/

int thrd_msgq_send(THRD_MSGQ msgq, struct listp_link *link_send)
{/*{{{*/
    if(!msgq || !link_send) return(RET_ERR_PARA);

    pthread_mutex_lock(&msgq->mutex);

    if(listp_is_empty(&msgq->head))
        pthread_cond_signal(&msgq->cond);
    listp_add_tail(link_send, &msgq->head);
    ++msgq->count;

    pthread_mutex_unlock(&msgq->mutex);

    return(0);
}/*}}}*/

int thrd_msgq_count(THRD_MSGQ msgq)
{/*{{{*/
    if(!msgq) return(RET_ERR_PARA);
    pthread_mutex_lock(&msgq->mutex);
    int count=msgq->count;
    pthread_mutex_unlock(&msgq->mutex);
    return(count);
}/*}}}*/
